package buaa.qel.rpc.server.core;

import buaa.qel.rpc.common.message.Beat;
import buaa.qel.rpc.common.message.RpcRequestMessage;
import buaa.qel.rpc.common.message.RpcResponseMessage;
import buaa.qel.rpc.common.util.ServiceKeyMaker;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import net.sf.cglib.reflect.FastClass;

import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class RpcRequestHandler extends SimpleChannelInboundHandler<RpcRequestMessage> {
    private final Map<String, Object> serviceMap;
    public final ThreadPoolExecutor requestHandlerThreadPool;

    public RpcRequestHandler(Map<String, Object> serviceMap, ThreadPoolExecutor threadPoolExecutor) {
        this.serviceMap = serviceMap;
        this.requestHandlerThreadPool = threadPoolExecutor;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, final RpcRequestMessage request) {
        if (Beat.BEAT_ID.equalsIgnoreCase(request.getSequenceId())) {
            log.info("Server read heartbeat ping");
            return;
        }
        requestHandlerThreadPool.execute(() -> {
            log.info("Receive request " + request.getSequenceId());
            RpcResponseMessage response = new RpcResponseMessage();
            response.setSequenceId(request.getSequenceId());
            try {
                Object result = handle(request);
                response.setReturnValue(result);
            } catch (Exception e) {
                response.setExceptionValue(e);
                log.error("RPC Server handle request error", e);
            }
            ctx.writeAndFlush(response).addListener(promise -> {
                log.info("Send response for request " + request.getSequenceId());
            });
        });
    }

    private Object handle(RpcRequestMessage request) throws Exception {
        String interfaceName = request.getInterfaceName();
        String serviceName = request.getServiceName();
        String serviceKey = ServiceKeyMaker.makeServiceKey(interfaceName, serviceName);
        Object serviceBean = serviceMap.get(serviceKey);
        if (serviceBean == null) {
            log.error("Can not find service implement with interfaceName: {} and serviceName: {}", interfaceName, serviceName);
            return null;
        }

        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameterValue = request.getParameterValue();

        log.debug(serviceClass.getName());
        log.debug(methodName);
        for (int i = 0; i < parameterTypes.length; i++) {
            log.debug(parameterTypes[i].getName());
        }
        for (int i = 0; i < parameterValue.length; i++) {
            log.debug(parameterValue[i].toString());
        }

        //cglib dynamic proxy
        FastClass serviceFastClass = FastClass.create(serviceClass);
        int methodIndex = serviceFastClass.getIndex(methodName, parameterTypes);
        return serviceFastClass.invoke(methodIndex, serviceBean, parameterValue);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.warn("Server caught exception: " + cause.getMessage());
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            ctx.channel().close();
            log.warn("Channel idle in last {} seconds, close it", Beat.BEAT_TIMEOUT);
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
